package net.wlm.realtime.funcation;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

public class CdcDataEtlMapFunction extends RichMapFunction<String,String> {

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    /*
     * 解析JSON
     * 提取字段值
     * 拼接字符串
     * 返回
     */

    @Override
    public String map(String value) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        // after值,插入
        String afterValue = jsonObject.getString("after");
        // before值,删除
        String beforeValue = jsonObject.getString("before");
        // op 值
        String opValue = jsonObject.getString("op");
        // ts_ms 值
        Long ts_msValue = jsonObject.getLong("ts_ms");
        // db 值, table 值
        JSONObject object = jsonObject.getJSONObject("source");
        String dbValue = object.getString("db");
        String tableValue = object.getString("table");

        String operatorValue;
        switch (opValue){
            case "u":
                operatorValue = "update";
                break;
            case "r":
            case "c":
                operatorValue = "insert";
                break;
            case "d":
                operatorValue = "delete";
                break;
            default:
                operatorValue = "unknown";
        }
        JSONObject newObject = new JSONObject();
        newObject.put("db_name",dbValue);
        newObject.put("table_name",tableValue);
        newObject.put("operate_type",operatorValue);
        newObject.put("ts_ms",ts_msValue);
        // data 是字符串
        if("d".equals(opValue)){
            newObject.put("operate_date",JSON.parseObject(beforeValue));
        }else {
            newObject.put("operate_date",JSON.parseObject(afterValue));
        }
        return newObject.toJSONString();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

}